package net.sina.realtime.traffic.job3;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import net.sina.realtime.traffic.bean.MonitorInfo;
import net.sina.realtime.traffic.schema.JSONDeserializationSchema;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class DangerousDriverController {

    // 静态限速缓存 - 解决序列化问题
    private static final Cache<String, Integer> SPEED_LIMIT_CACHE = CacheBuilder.newBuilder()
            .maximumSize(1000)
            .expireAfterWrite(1, TimeUnit.HOURS)
            .build();

    // 静态限速服务 - 解决序列化问题
    private static class SpeedLimitService {
        public int getSpeedLimit(String monitorId) {
            // 实际项目中应查询数据库或外部服务
            if (monitorId == null) return 60;

            if (monitorId.startsWith("highway")) {
                return 120; // 高速公路限速120
            } else if (monitorId.startsWith("urban")) {
                return 80;  // 城市快速路限速80
            } else if (monitorId.startsWith("road")) {
                return 60;  // 普通道路限速60
            } else {
                return 60;  // 默认限速60
            }
        }
    }

    // 可序列化的超速条件类 - 解决序列化问题
    private static class OverspeedCondition extends SimpleCondition<MonitorInfo> {
        private final SpeedLimitService speedLimitService = new SpeedLimitService();

        @Override
        public boolean filter(MonitorInfo value) throws Exception {
            if (value.getMonitorId() == null || value.getMonitorId().isEmpty()) {
                return false;
            }

            // 从缓存获取限速值，如果不存在则查询服务
            Integer limitSpeed = SPEED_LIMIT_CACHE.get(value.getMonitorId(),
                    () -> speedLimitService.getSpeedLimit(value.getMonitorId()));

            return value.getSpeed() > limitSpeed * 1.2;
        }
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers", "node101:9092");
        prop.setProperty("group.id", "car-group2");
        FlinkKafkaConsumer<MonitorInfo> consumer =
                new FlinkKafkaConsumer<>("topic-wx",
                        new JSONDeserializationSchema<>(MonitorInfo.class), prop);

        DataStreamSource<MonitorInfo> ds1 = env.addSource(consumer);
        ds1.print("原始数据:");

        SingleOutputStreamOperator<MonitorInfo> ds2 = ds1
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<MonitorInfo>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(
                                (SerializableTimestampAssigner<MonitorInfo>)
                                        (element, recordTimestamp) -> element.getActionTime()
                        ));

        // 使用可序列化的超速条件类
        OverspeedCondition overspeedCondition = new OverspeedCondition();

        // 定义CEP模式：10分钟内任意3次超速
        Pattern<MonitorInfo, ?> pattern = Pattern.<MonitorInfo>begin("first", AfterMatchSkipStrategy.skipPastLastEvent())
                .where(overspeedCondition)
                .followedBy("second").where(overspeedCondition)
                .followedBy("third").where(overspeedCondition)
                .within(Time.minutes(10));

        // 在数据流上应用模式匹配
        PatternStream<MonitorInfo> patternStream = CEP.pattern(
                ds2.keyBy(MonitorInfo::getCar),
                pattern
        );

        // 输出匹配结果
        SingleOutputStreamOperator<String> ds3 = patternStream.select(
                new PatternSelectFunction<MonitorInfo, String>() {
                    @Override
                    public String select(Map<String, List<MonitorInfo>> patternMap) throws Exception {
                        List<MonitorInfo> first = patternMap.get("first");
                        List<MonitorInfo> second = patternMap.get("second");
                        List<MonitorInfo> third = patternMap.get("third");

                        StringBuilder sb = new StringBuilder("\n===== 危险驾驶警报 =====");
                        sb.append("\n车牌号: ").append(first.get(0).getCar());
                        sb.append("\n超速事件1: [卡口:").append(first.get(0).getMonitorId())
                                .append(", 速度:").append(first.get(0).getSpeed())
                                .append(", 时间:").append(first.get(0).getActionTime()).append("]");
                        sb.append("\n超速事件2: [卡口:").append(second.get(0).getMonitorId())
                                .append(", 速度:").append(second.get(0).getSpeed())
                                .append(", 时间:").append(second.get(0).getActionTime()).append("]");
                        sb.append("\n超速事件3: [卡口:").append(third.get(0).getMonitorId())
                                .append(", 速度:").append(third.get(0).getSpeed())
                                .append(", 时间:").append(third.get(0).getActionTime()).append("]");
                        sb.append("\n======================");
                        return sb.toString();
                    }
                }
        );

        ds3.print("危险驾驶车辆:");

        env.execute("DangerousDriverDetection");
    }
}